package com.sk.rpcAssistNetty.io.netty.client;

import com.sk.rpcAssistNetty.domain.ProvideInfoValue;
import com.sk.rpcAssistNetty.io.netty.RpcRequest;
import com.sk.rpcAssistNetty.io.netty.RpcResponse;
import com.sk.rpcAssistNetty.io.netty.codec.RpcDecoder;
import com.sk.rpcAssistNetty.io.netty.codec.RpcEncoder;
import com.sk.rpcAssistNetty.io.netty.server.NettyServer;
import com.sk.rpcAssistNetty.io.protocol.HessianSerialize;
import com.sk.rpcAssistNetty.io.protocol.JsonSerialization;
import com.sk.rpcAssistNetty.io.protocol.SerializeProtocol;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
import org.springframework.core.io.ClassPathResource;

import java.util.Date;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;


/**
 * description：netty客户端
 * ①：connect连接netty服务器
 * ②：重试功能
 * ③：实现发送消息的接口
 * ④：close功能
 *
 * @author 孙文斌
 * @Date: 2021/3/22 14:41
 */
@Slf4j
public class NettyClient {

    private String host;
    private Integer port;
    private Channel channel;
    private ClientHandler clientHandler;
    private SerializeProtocol serializeProtocol;
    private ProvideInfoValue provideInfo;
    private EventLoopGroup eventLoopGroup;
    //最大重试次数
    private static final int RETRY_MAX = 5;

    public NettyClient(ProvideInfoValue provideInfo,SerializeProtocol serializeProtocol) {
        this.serializeProtocol = serializeProtocol;
        this.provideInfo = provideInfo;
        this.host = provideInfo.getIp();
        this.port = provideInfo.getPort();
    }

    public static void main(String[] args) {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        ClientHandler clientHandler = new ClientHandler();
        Bootstrap bootstrap = new Bootstrap();
        SerializeProtocol serializeProtocol = HessianSerialize.getInstance();
        bootstrap.group(eventLoopGroup)
                //指定传输使用的Channel
                .channel(NioSocketChannel.class)
//                .option(ChannelOption.SO_KEEPALIVE, true)
//                .option(ChannelOption.TCP_NODELAY, true)
//                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new LengthFieldBasedFrameDecoder(65535, 0, 4));
                        //添加编码器
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, serializeProtocol));
                        //添加解码器
                        pipeline.addLast(new RpcDecoder(RpcResponse.class, serializeProtocol));
                        //请求处理类
                        pipeline.addLast(clientHandler);
                    }
                });
//        connect(bootstrap, host, port, RETRY_MAX);
        connect0(bootstrap, "127.0.0.1", 18089, RETRY_MAX);
    }
    private static void connect0(Bootstrap bootstrap, String host, int port, int retry) {
        ChannelFuture channelFuture = bootstrap.connect(host, port).addListener(future -> {
            if (future.isSuccess()) {
                log.info("netty server connect success!");
            } else if (retry == 0) {
                log.error("alread retry conect {} times,connect fail", RETRY_MAX);
            } else {
                //当前重试的次数
                int order = RETRY_MAX - retry + 1;
                //延迟重试时间间隔
                int delay = 1 << order;
                log.error("{} : Connection failed，retry connect times is {}....", new Date(), order);
                //() -> connect(bootstrap, host, port, retry - 1)是函数式编程的形式，其实是对runnable方法的简写
                bootstrap.config().group().schedule(() -> connect0(bootstrap, host, port, retry - 1), delay, TimeUnit.SECONDS);
            }
        });
        Channel channel = channelFuture.channel();
    }

    /**
     * description：初始化netty client参数配置封装启动助手
     * @Date: 2021/3/22 14:41
     */
    public void initClient(){
        eventLoopGroup = new NioEventLoopGroup();
        clientHandler = new ClientHandler();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                //指定传输使用的Channel
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true) //长链接的时候保活设置
                .option(ChannelOption.TCP_NODELAY, true) //低延迟设置，禁用nagle算法
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)//连接超时设置
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        //添加编码器
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, serializeProtocol));
                        //添加解码器
                        pipeline.addLast(new RpcDecoder(RpcResponse.class, serializeProtocol));
                        //请求处理类
                        pipeline.addLast(clientHandler);
                    }
                });
        connect(bootstrap, host, port, RETRY_MAX);
//        connect(bootstrap, host, 38885, RETRY_MAX);
    }

    /**
     * description：初始化netty client，并且指定重试次数【Future-Listener异步模型】
     * Bind、Write、Connect 等操作会简单的返回一个 ChannelFuture，通过此获取操作状态
     *     通过 isDone 方法来判断当前操作是否完成；
     *     通过 isSuccess 方法来判断已完成的当前操作是否成功；
     *     通过 getCause 方法来获取已完成的当前操作失败的原因；
     *     通过 isCancelled 方法来判断已完成的当前操作是否被取消
     * 【通过 addListener 方法来注册监听器，当操作已完成(isDone 方法返回完成)，将会通知指定的监听器；如果Future对象已完成，则通知指定的监听器】
     * @Date: 2021/3/22 14:41
     */
    private void connect(Bootstrap bootstrap, String host, int port, int retry) {
        ChannelFuture channelFuture = bootstrap.connect(host, port).addListener(future -> {
            if (future.isSuccess()) {
                log.info("netty server connect success!");
                log.info("连接netty服务端成功!host={},prot={}",host,port);
            } else if (retry == 0) {
                log.error("alread retry conect {} times,connect fail", RETRY_MAX);
                log.error("重试次数已用完，放弃连接");
            } else {
                //当前重试的次数
                int order = RETRY_MAX - retry + 1;
                //延迟重试时间间隔
                int delay = 1 << order;
                log.error("{} : 连接失败，第 {} 重连....", new Date(), order);
                log.error("{} : Connection failed，retry connect times is {}....", new Date(), order);
                //() -> connect(bootstrap, host, port, retry - 1)是函数式编程的形式，其实是对runnable方法的简写
                bootstrap.config().group().schedule(() -> connect(bootstrap, host, port, retry - 1), delay, TimeUnit.SECONDS);
            }
        });
        try {
            //todo：这里使用同步的get方法获取连接结果，因为使用异步可能会出现连接还没有建立好，就发送请求获取结果，导致空指针异常
            channelFuture.get();
        } catch (Exception e) {
            e.printStackTrace();
            log.error("同步连接netty服务端异常");
        }
        channel = channelFuture.channel();
    }

    /**
     * description：发送请求，并且异步（wait-notify机制）获取结果
     * @Date: 2021/3/22 14:41
     */
    public RpcResponse sendMsg(final RpcRequest request){
        try {
            //tip:注意，这里的await是ChannelFuture自带的，它不需要被外部唤醒，当写出操作完成（包括成功和失败、异常、超时）就是被唤醒
            //这里之所以需要添加写操作的await，就是想让写操作完成后（不管成功或者失败）再去接收响应，保证逻辑的正确
            channel.writeAndFlush(request).await(); //如果停掉server端，会返回如下情况
        } catch (InterruptedException e) {
            e.printStackTrace();
            log.error("send message fail,request info is {},exception message is {}",request,e.getMessage());
        }
        try {
            //在这里包装方法中使用了DefaultFuture才是发挥【等待\通知】机制
            return clientHandler.getRpcResponse(request.getRequestId());
        } finally {
            //todo：优化：每次请求完毕，销毁netty客户端
            eventLoopGroup.shutdownGracefully();
            channel.closeFuture().syncUninterruptibly();
        }
    }
}
